package com.zs.dbmonitor.config;

import com.zs.dbmonitor.model.ChangeDataModel;
import io.debezium.connector.mysql.MySqlConnector;
import io.debezium.data.Envelope;
import io.debezium.embedded.Connect;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.RecordChangeEvent;
import io.debezium.engine.format.ChangeEventFormat;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;

import static io.debezium.data.Envelope.FieldName.*;
import static java.util.stream.Collectors.toMap;

/**
 * @author ZhaoShuai
 * @version 1.0
 * @description: TODO
 * @date 2024/6/20 21:49
 */
@Component
@ConfigurationProperties(prefix = "debezium.datasource.mysql")
@Data
@Slf4j
public class DebeziumConfiguration {

    private String offsetStorage;


    private String offsetStorageFilePath;


    private String offsetFlushInterval;


    private String connectorName;


    private String connectorClass;


    private String databaseHost;


    private String databasePort;


    private String databaseUser;


    private String databasePassword;


    private String databaseIncludeList;


    private String includeSchemaChanges;


    private String databaseServerId;


    private String databaseServerName;


    private String databaseHistory;


    private String databaseHistoryFilePath;

    /**
     * Debezium 配置.
     *
     * @return configuration
     */
    @Bean
    public io.debezium.config.Configuration debeziumConfig() {
        return io.debezium.config.Configuration.create()
                //            连接器的Java类名称
                .with("connector.class", connectorClass)
                //            偏移量持久化，用来容错 默认值
                .with("offset.storage", offsetStorage)
                //                偏移量持久化文件路径 默认/tmp/offsets.dat  如果路径配置不正确可能导致无法存储偏移量 可能会导致重复消费变更
                //                如果连接器重新启动，它将使用最后记录的偏移量来知道它应该恢复读取源信息中的哪个位置。
                .with("offset.storage.file.filename", offsetStorageFilePath)
                //                捕获偏移量的周期
                .with("offset.flush.interval.ms", offsetFlushInterval)
                //               连接器的唯一名称
                .with("name", connectorName)
                //                数据库的hostname
                .with("database.hostname", databaseHost)
                //                端口
                .with("database.port", databasePort)
                //                用户名
                .with("database.user", databaseUser)
                //                密码
                .with("database.password", databasePassword)
                //                 包含的数据库列表
                .with("database.include.list", databaseIncludeList)
                //                是否包含数据库表结构层面的变更，建议使用默认值true
                .with("include.schema.changes", includeSchemaChanges)
                //                mysql.cnf 配置的 server-id
                .with("database.server.id", databaseServerId)
                //                 MySQL 服务器或集群的逻辑名称
                .with("database.server.name", databaseServerName)
                //                历史变更记录
                .with("database.history", databaseHistory)
                //                历史变更记录存储位置
                .with("database.history.file.filename", databaseHistoryFilePath)
                .build();
    }

    /**
     * Debezium server bootstrap debezium server bootstrap.
     *
     * @param configuration the configuration
     * @return the debezium server bootstrap
     */
    @Bean
    DebeziumServerBootstrap debeziumServerBootstrap(io.debezium.config.Configuration configuration) {
        DebeziumServerBootstrap debeziumServerBootstrap = new DebeziumServerBootstrap();
        DebeziumEngine<RecordChangeEvent<SourceRecord>> debeziumEngine = DebeziumEngine.create(ChangeEventFormat.of(Connect.class))
                .using(configuration.asProperties())
                .notifying(this::handlePayload)
                .build();

        debeziumServerBootstrap.setDebeziumEngine(debeziumEngine);
        return debeziumServerBootstrap;
    }


    private void handlePayload(List<RecordChangeEvent<SourceRecord>> recordChangeEvents, DebeziumEngine.RecordCommitter<RecordChangeEvent<SourceRecord>> recordCommitter) {
        recordChangeEvents.forEach(r -> {
            SourceRecord sourceRecord = r.record();
            Struct sourceRecordChangeValue = (Struct) sourceRecord.value();
            String changeTable = sourceRecord.topic().substring(sourceRecord.topic().lastIndexOf(".") + 1);
            ChangeDataModel dataModel = new ChangeDataModel();
            dataModel.setTable(changeTable);
            if (sourceRecordChangeValue != null) {
                // 判断操作的类型 过滤掉读 只处理增删改   这个其实可以在配置中设置
                Envelope.Operation operation = Envelope.Operation.forCode((String) sourceRecordChangeValue.get(OPERATION));
                if ((operation == Envelope.Operation.UPDATE)) {
                    Map<String, Object> beforeChangeData = getPayloadData(sourceRecordChangeValue, BEFORE);
                    Map<String, Object> afterChangeData = getPayloadData(sourceRecordChangeValue, AFTER);
                    dataModel.setOperationType(operation.name());
                    dataModel.setBefore(beforeChangeData);
                    dataModel.setAfter(afterChangeData);
                } else if (operation == Envelope.Operation.CREATE) {
                    Map<String, Object> afterChangeData = getPayloadData(sourceRecordChangeValue, AFTER);
                    dataModel.setOperationType(operation.name());
                    dataModel.setAfter(afterChangeData);
                } else if (operation == Envelope.Operation.DELETE) {
                    Map<String, Object> beforeChangeData = getPayloadData(sourceRecordChangeValue, BEFORE);
                    dataModel.setOperationType(operation.name());
                    dataModel.setBefore(beforeChangeData);
                }
                log.info(dataModel.toString());
            }
        });
    }

    private Map<String, Object> getPayloadData(Struct sourceRecordChangeValue, String record) {
        // 获取增删改对应的结构体数据
        Struct struct = (Struct) sourceRecordChangeValue.get(record);
        // 将变更的行封装为Map
        return struct.schema().fields().stream()
                .map(Field::name)
                .filter(fieldName -> struct.get(fieldName) != null)
                .map(fieldName -> Pair.of(fieldName, struct.get(fieldName)))
                .collect(toMap(Pair::getKey, Pair::getValue));
    }
}
